package org.etlt.job;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.etlt.EtltRuntimeException;
import org.etlt.check.CheckList;
import org.etlt.load.Loader;
import org.etlt.util.SystemUtil;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ParalleledJobExecutor {

    private int count = SystemUtil.getCpuCount();

    protected Log log = LogFactory.getLog(getClass());

    public ParalleledJobExecutor() {

    }

    public ParalleledJobExecutor(int count) {
        if (count > 0 && count <= SystemUtil.getCpuCount())
            this.count = count;
    }

    public void execute(File jobDirectory) {
        try {
            final JobContext jobContext = new JobContext(jobDirectory);
            jobContext.init();
            execute(jobContext);
        } catch (IOException e) {
            throw new EtltRuntimeException(e);
        }

    }

    public void execute(JobContext context) {
        log.info("start job executing:" + context.getJobSetting().getName());
        executeLoaders(context);
        executeChecklists(context);
        log.info("job executing finished:" + context.getJobSetting().getName());
        log.info("job execution: " + context.getJobExecution());
    }

    /**
     * @param jobContext
     */
    protected void executeLoaders(JobContext jobContext) {
        int parallel = jobContext.getJobSetting().getParallel();
        final JobExecutor executor = new JobExecutor();
        log.info("There are " + jobContext.getAllLoaders().size() + " loader(s) inited.");
        for (List<Loader> batchLoaders = jobContext.getUnfinishedLoaders();
             !batchLoaders.isEmpty();
             batchLoaders = jobContext.getUnfinishedLoaders()) {
            parallel = parallel > this.count ? parallel : this.count;
            log.info("There are unfinished " + batchLoaders.size() + " loaders which be started in " + parallel + " threads.");
            this.executionLoaders.addAll(batchLoaders);
            final ExecutorService executorService = Executors.newFixedThreadPool(parallel, new NamingThreadFactory("loader-executors"));
            for (Loader loader = getLoader(); loader != null; loader = getLoader()) {
                final Loader taskLoader = loader;
                executorService.execute(() -> {
                    executor.execute(jobContext, taskLoader);
                });
            }
            try {
                executorService.shutdown();
                executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                this.executionLoaders.clear();
            }
        }
        log.info("all loaders executing finished.");
    }

    protected void executeChecklists(JobContext jobContext) {
        List<CheckList> checkLists = jobContext.getAllChecklists();
        log.info("there are " + checkLists.size() + " checklists which be started in " + this.count + " threads.");
        this.executionChecklists.addAll(checkLists);
        JobExecutor executor = new JobExecutor();
        final ExecutorService executorService = Executors.newFixedThreadPool(this.count, new NamingThreadFactory("checklist-executors"));
        for (CheckList checkList = getChecklist(); checkList != null; checkList = getChecklist()) {
            final CheckList todoChecklist = checkList;
            executorService.execute(() -> {
                executor.execute(jobContext, todoChecklist);
            });
        }
        try {
            executorService.shutdown();
            executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            this.executionChecklists.clear();
        }
        log.info("all checklists executing finished.");
    }

    protected Loader getLoader() {
        return this.executionLoaders.poll();
    }

    private Queue<Loader> executionLoaders = new ConcurrentLinkedQueue<>();

    protected CheckList getChecklist() {
        return this.executionChecklists.poll();
    }

    private Queue<CheckList> executionChecklists = new ConcurrentLinkedQueue<>();

    /**
     * thread factory that doesn't use the default naming strategy, using {{@link #name}} as group name, the thread name is {{@link #name}}-sequence, the sequence begins from 1 (included)
     */
    private static class NamingThreadFactory implements ThreadFactory {

        private final String name;

        private final AtomicInteger counter = new AtomicInteger(1);

        private NamingThreadFactory(String name) {
            this.name = name;
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName(this.name + "-" + counter.getAndIncrement());
            return thread;
        }
    }
}
